package com.cqwshzj.thread.worker;/**
 * @Description:
 * @Auther: Huang Zhijin
 * @Version: 1.0.0
 * @Date: 2021/4/12 10:40
 */

import com.cqwshzj.common.JobConstant;
import com.cqwshzj.thread.bean.Stateable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 *
 * 〈〉
 *
 * @author Huang Zhijin
 * @create 2021/4/12
 * @since 1.0.0
 */
public class MainCheckWorker<T extends Stateable> implements Runnable {

	private static final Logger log = LoggerFactory.getLogger(MainCheckWorker.class);

	//-----------

	/** 下载队列 */
	private LinkedBlockingQueue<T> downloadQueue;
	/** 校验队列 */
	private LinkedBlockingQueue<T> checkQueue;
	/** 下载出错计数 */
	private ConcurrentHashMap<String,String> checkFailedMap;

	/** 主下载线程和校验线程的 计数器 */
	private CountDownLatch mainWorkerSignal;

	//-----------
	/** 校验统计 */
	private int checkTimeCount = 0;
	/** 校验通过统计 */
	private int checkSuccessCount = 0;

	//-----------

	public MainCheckWorker(LinkedBlockingQueue<T> downloadQueue,
						   LinkedBlockingQueue<T> checkQueue,
						   CountDownLatch mainWorkerSignal ) {
		this.downloadQueue = downloadQueue;
		this.checkQueue = checkQueue;
		this.mainWorkerSignal = mainWorkerSignal;
		this.checkFailedMap = new ConcurrentHashMap<>();
	}


	//-----------

	@Override
	public void run() {
		log.info("--------------- check begin ---------------");
		try {
			long start = System.currentTimeMillis();
			long waitedTime = 0L;
			int sleepPeriod = 50;
			int logPeriod = sleepPeriod * 100;

			while ( 1 != downloadQueue.size() + checkQueue.size() + mainWorkerSignal.getCount() ){
				if ( 0 == checkQueue.size() ){
					try {
						TimeUnit.MILLISECONDS.sleep( sleepPeriod );
						waitedTime += sleepPeriod;
						if ( waitedTime % logPeriod == 0 ){
							if (log.isDebugEnabled()) {
								log.debug(" downloadQueue=[{}], checkQueue=[{}], threads=[{}] ",downloadQueue.size(), checkQueue.size(), mainWorkerSignal.getCount() );
							}
						}
					}catch (InterruptedException e){
						log.error(e.getMessage(),e);
						return;
					}
					continue;
				}

				checkSubject( checkQueue.peek() );
			}
			log.info(" all subjects check OK! success=[true], cost=[\t{}]ms, thread wait=[\t{}]ms. ",(System.currentTimeMillis() -start), waitedTime);
		}catch (Throwable e){
			log.error(e.getMessage(),e);
		}
		//表示当前线程执行关闭，剩余主要的工作线程总数计数器，计数器减一
		mainWorkerSignal.countDown();
		// other business
		beforeFinishCheck();
		log.info("--------------- check end ---------------");
	}


	private void beforeFinishCheck() {
		log.debug(" beforeFinishCheck");
	}


	private void checkSubject(T subject) {
		checkTimeCount++;
		if ( !JobConstant.COMMON_YES.equals( subject.getState() )  ){
			//校验未通过
			handleCheckFailed( subject );
		}else{
			//校验通过

			checkFailedMap.remove( String.valueOf(subject.getIdentityField()) );
			checkQueue.remove( subject );
			checkSuccessCount++;
			log.info(" check Subject OK! . id=[{}] state=[{}] .", subject.getIdentityField(), subject.getState());
		}

	}


	private void handleCheckFailed(T subject) {
		String key = String.valueOf( subject.getIdentityField() );
		//首次校验失败
		if ( !checkFailedMap.containsKey(key) ){
			checkFailedMap.put( key , "1");
			downloadQueue.offer( subject );
			checkQueue.remove( subject );
			return;
		}

		//曾经校验失败
		String failedCountStr = checkFailedMap.get( key );
		int failedCount = Integer.parseInt( failedCountStr );
		int retryMaxTime = 5;
		if ( failedCount > retryMaxTime ){
			log.error(" check failed. reach max retry time . id=[{}] checkFailedCount=[{}] .", key, failedCount );
			// other business code
			// 从校验队列中移除
			checkFailedMap.remove(key);
			checkQueue.remove(key);
			return;
		}

		failedCount++;
		if (failedCount == retryMaxTime) {
			log.error(" check failed for {} time. id=[{}].", failedCount, key );
		}

		//移除校验队列
		checkQueue.remove(key);
		checkFailedMap.put( key, String.valueOf(failedCount) );
		downloadQueue.offer( subject );
	}
}
